package com.hksj.mqtt.callback;

import com.hksj.mqtt.annotation.MqttConfiguration;
import com.hksj.mqtt.annotation.MqttSubscription;
import com.hksj.mqtt.client.IMqttClient;
import com.hksj.mqtt.listener.MqttListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.PostConstruct;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


public class DefaultMqttCallBackAdapter implements  MqttCallBackAdapter,ApplicationContextAware {

    private ApplicationContext applicationContext;

    private ThreadPoolTaskExecutor executor;

    private ConcurrentHashMap<String, List<MqttListener>> subscriptionCache = new ConcurrentHashMap<>();

    private ConcurrentHashMap<String, List<MqttSubscriptionExecutor>> subscriptionMethodCache = new ConcurrentHashMap<>();

    private ConcurrentHashMap<String,Integer> topicQosCache = new ConcurrentHashMap<>();

    private Set<String> topicsCache = new HashSet<>();

    private String targetClient = "default";

    public DefaultMqttCallBackAdapter(ThreadPoolTaskExecutor executor) {
        this.executor = executor;
    }

    public DefaultMqttCallBackAdapter(ThreadPoolTaskExecutor executor,String targetClient) {
        this.executor = executor;
        this.targetClient = targetClient;
    }

    @PostConstruct
    public void init(){
        //接口
        Map<String, MqttListener> mqttListenerMap = applicationContext.getBeansOfType(MqttListener.class);
        for (MqttListener mqttListener : mqttListenerMap.values()) {
            MqttSubscription subscription = mqttListener.getClass().getAnnotation(MqttSubscription.class);
            if (subscription != null && targetClient.equals(subscription.targetClient())){
                String[] topics = subscription.topics();
                int qos = subscription.qos();
                for (String topic : topics) {
                    topicsCache.add(topic);

                    List<MqttListener> currentTopicSubscriptions = subscriptionCache.getOrDefault(topic, new ArrayList<>());
                    currentTopicSubscriptions.add(mqttListener);
                    subscriptionCache.put(topic,currentTopicSubscriptions);

                    if (qos >= topicQosCache.getOrDefault(topic,0)){
                        topicQosCache.put(topic,qos);
                    }
                }
            }
        }

        //方法
        Map<String, Object> configurationMap = applicationContext.getBeansWithAnnotation(MqttConfiguration.class);
        for (Object mqttConfiguration : configurationMap.values()) {
            System.out.println(mqttConfiguration);
            for (Method declaredMethod : mqttConfiguration.getClass().getDeclaredMethods()) {
                if (declaredMethod.isAnnotationPresent(MqttSubscription.class)){
                    MqttSubscription subscription = declaredMethod.getAnnotation(MqttSubscription.class);
                    if (subscription != null && targetClient.equals(subscription.targetClient())){
                        String[] topics = subscription.topics();
                        int qos = subscription.qos();
                        for (String topic : topics) {
                            topicsCache.add(topic);

                            List<MqttSubscriptionExecutor> currentSubscriptionExecutors = subscriptionMethodCache.getOrDefault(topic, new ArrayList<>());
                            currentSubscriptionExecutors.add(new MqttSubscriptionExecutor(mqttConfiguration, declaredMethod));
                            subscriptionMethodCache.put(topic,currentSubscriptionExecutors);

                            if (qos >= topicQosCache.getOrDefault(topic,0)){
                                topicQosCache.put(topic,qos);
                            }
                        }
                    }
                }
            }
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void initSubscript(IMqttClient mqttClient) {
        executor.execute(() -> {
            Set<String> toDeleteTopics = new HashSet<>();
            for (String topic : topicsCache) {
                for (String regex : topicsCache) {
                    if (topic != regex){
                        if (topic.matches(topToRegex(regex))){
                            toDeleteTopics.add(topic);
                        }
                    }
                }
            }
            for (String topic : topicsCache) {
                if (toDeleteTopics.contains(topic)){
                    continue;
                }
                mqttClient.subscribe(topic,topicQosCache.get(topic));
            }
        });
    }

    @Override
    public void handleMsg(String topic, MqttMessage mqttMessage) {
        executor.execute(() -> {
            try{
                for (String subscribedTopic : topicsCache) {
                    if (matches(topic,subscribedTopic)){
                        for (MqttListener mqttListener : subscriptionCache.getOrDefault(subscribedTopic,new ArrayList<>())) {
                            mqttListener.receive(topic,mqttMessage);
                        }

                        for (MqttSubscriptionExecutor mqttSubscriptionExecutor : subscriptionMethodCache.getOrDefault(subscribedTopic, new ArrayList<>())) {
                            mqttSubscriptionExecutor.execute(topic,mqttMessage);
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }

    private boolean matches(String str,String regex){
        if (str == null || "".equals(str)){
            return false;
        }
        if (str.startsWith("$SYS")){
            if (!regex.startsWith("$SYS")){
                return false;
            }
            regex = regex.replace("$","^\\$");
        }
        regex = regex.replaceAll("\\+","[^/]*").replaceAll("#",".*");
        return str.matches(regex);
    }

    private String topToRegex(String regex){
        return regex.replaceAll("\\+","[^/]*")
                .replaceAll("#",".*")
                .replace("$","^\\$");
    }

    static class MqttSubscriptionExecutor {
        public MqttSubscriptionExecutor(Object obj, Method method) {
            this.obj = obj;
            this.method = method;
        }

        private Object obj;

        private Method method;

        public void execute(Object... args){
            try {
                method.invoke(obj,args);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    @Override
    public ThreadPoolTaskExecutor getExecutor() {
        return executor;
    }

    public void setExecutor(ThreadPoolTaskExecutor executor) {
        this.executor = executor;
    }
}
